package com.shujia.flink.core;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import javax.xml.crypto.Data;

public class Demo6Car {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //读取数据
        DataStream<String> linesDS = env.readTextFile("flink/data/cars_sample.txt");

        //解析数据,将每一行转换成java对象
        DataStream<Car> cars = linesDS.map(line -> {
            String[] split = line.split(",");
            //将时间戳转换成毫秒级别
            long time = Long.parseLong(split[7]) * 1000;
            double speed = Double.parseDouble(split[8]);
            //构建java对象返回
            return new Car(split[0], split[1], split[2], split[3], split[4], split[5], split[6], time, speed);
        });
        /*
         * 需求
         * 1、实时统计道路的拥堵情况
         * 通过平均车速和车流量判断
         * 计算最近15分钟的过车数据进行计算，每1分钟计算一次
         */

        //设置时间字段和水位线
        DataStream<Car> assCarsDS = cars.assignTimestampsAndWatermarks(
                WatermarkStrategy
                        .<Car>forMonotonousTimestamps()
                        .withTimestampAssigner((car, ts) -> car.getTime())
        );

        //安装道路分组
        KeyedStream<Car, String> keyByDS = assCarsDS.keyBy(car -> car.getRoadId());

        //划分窗口
        WindowedStream<Car, String, TimeWindow> windowDS = keyByDS
                .window(SlidingEventTimeWindows.of(Time.minutes(15), Time.minutes(1)));

        //通过平均车速和车流量判断
        //process算子时flink底层算子，可以实现上层所有算子的功能
        DataStream<Result> resultsDS = windowDS.process(new ProcessWindowFunction<Car, Result, String, TimeWindow>() {
            /**
             * process方法每一个窗口执行一次
             * @param key ：分组的key(道路编号)
             * @param context ：flink上下文对象。可以或flink执行的环境信息
             * @param cars : 同一个窗口内所有的车辆数据
             * @param out：用于将数据发送到下游
             */
            @Override
            public void process(String key,
                                ProcessWindowFunction<Car, Result, String, TimeWindow>.Context context,
                                Iterable<Car> cars,
                                Collector<Result> out) throws Exception {
                //通过平均车速和车流量判断

                long flow = 0;
                Double sumSpeed = 0.0;
                for (Car car : cars) {
                    flow++;
                    sumSpeed += car.getSpeed();
                }
                //计算平均车速
                double avgSpeed = sumSpeed / flow;

                //将计算结果发送到下游
                out.collect(new Result(key, flow, avgSpeed));
            }
        });

        resultsDS.print();

        env.execute();

    }
}

class Result {
    private String roadId;
    private Long flow;
    private Double avgSpeed;

    @Override
    public String toString() {
        return "Result{" +
                "roadId='" + roadId + '\'' +
                ", flow=" + flow +
                ", avgSpeed=" + avgSpeed +
                '}';
    }

    public String getRoadId() {
        return roadId;
    }

    public void setRoadId(String roadId) {
        this.roadId = roadId;
    }

    public Long getFlow() {
        return flow;
    }

    public void setFlow(Long flow) {
        this.flow = flow;
    }

    public Double getAvgSpeed() {
        return avgSpeed;
    }

    public void setAvgSpeed(Double avgSpeed) {
        this.avgSpeed = avgSpeed;
    }

    public Result(String roadId, Long flow, Double avgSpeed) {
        this.roadId = roadId;
        this.flow = flow;
        this.avgSpeed = avgSpeed;
    }
}
